package common

import (
	"errors"
	"gitee.com/kinwyb/appTools/log"
	"github.com/nats-io/nats.go"
	"github.com/sirupsen/logrus"
	"strings"
	"time"
)

// NatsSubInfo 话题订阅信息
type NatsSubInfo struct {
	Topic        string `description:"话题"`
	Queue        string `description:"队列名称"`
	MsgHander    nats.MsgHandler
	subscription *nats.Subscription
}

type Nats struct {
	address    []string               //nats地址
	username   string                 //nats连接账户
	password   string                 //nats连接密码
	prefix     string                 //topic名称前缀
	isChange   bool                   //连接信息变化
	nc         *nats.Conn             //连接
	clientSubs map[string]NatsSubInfo //客户端
	log        *logrus.Entry          //日志
}

func NewNats(address []string, username string, password string, prefix string) (*Nats, error) {
	ret := &Nats{
		address:    address,
		username:   strings.TrimSpace(username),
		password:   password,
		prefix:     strings.TrimSpace(prefix),
		clientSubs: map[string]NatsSubInfo{},
	}
	err := ret.Init(false)
	go ret.checkSubState()
	return ret, err
}

func (n *Nats) Init(reload bool) error {
	if !reload && n.nc != nil && !n.isChange {
		return nil
	}
	n.Close()
	options := nats.GetDefaultOptions()
	options.Servers = n.address
	options.User = n.username
	options.Password = n.password
	nc, err := options.Connect()
	if err != nil {
		return err
	}
	n.nc = nc
	clientHandlers := n.clientSubs
	for _, v := range clientHandlers {
		n.NewSubscribe(v)
	}
	return nil
}

func (n *Nats) Reload() {
	n.Init(true)
}

// Put 推送一个消息
func (n *Nats) Put(topic string, msg []byte) error {
	n.Init(false)
	return n.nc.Publish(n.prefix+topic, msg)
}

// Close 关闭
func (n *Nats) Close() {
	if n.nc != nil {
		n.nc.Close()
	}
	for _, v := range n.clientSubs {
		if v.subscription != nil {
			v.subscription.Unsubscribe()
			v.subscription = nil
		}
	}
}

// NewSubscribe 创建客户端
func (n *Nats) NewSubscribe(subInfo NatsSubInfo) error {
	n.Init(false)
	if n.nc == nil {
		return errors.New("nats未连接")
	} else if subInfo.Topic == "" {
		return errors.New("话题不能为空")
	} else if subInfo.MsgHander == nil {
		return errors.New("MsgHandler不能为空")
	}
	var err error
	if subInfo.Queue != "" {
		subInfo.subscription, err = n.nc.QueueSubscribe(n.prefix+subInfo.Topic,
			subInfo.Queue, subInfo.MsgHander)
	} else {
		subInfo.subscription, err = n.nc.Subscribe(n.prefix+subInfo.Topic, subInfo.MsgHander)
	}
	if err != nil {
		return err
	}
	n.clientSubs[subInfo.Topic+subInfo.Queue] = subInfo
	return nil
}

// SetAddress 设置地址,
func (n *Nats) SetAddress(address []string) {
	str := strings.TrimSpace(strings.Join(address, ","))
	oldStr := strings.TrimSpace(strings.Join(n.address, ","))
	if str == oldStr {
		return
	}
	n.address = address
	n.isChange = true
}

func (n *Nats) SetUsername(username string) {
	username = strings.TrimSpace(username)
	if n.username == username {
		return
	}
	n.username = username
	n.isChange = true
}

func (n *Nats) SetPassword(password string) {
	if n.password == password {
		return
	}
	n.password = password
	n.isChange = true
}

func (n *Nats) SetPrefix(prefix string) {
	prefix = strings.TrimSpace(prefix)
	if n.prefix == prefix {
		return
	}
	n.prefix = prefix
	n.isChange = true
}

// 检验sub是否还是在线
func (n *Nats) checkSubState() {
	defer func() {
		if err := recover(); err != nil {
			log.AppTool.WithField("error", err).Error("nats sub check painc")
			go n.checkSubState()
		}
	}()
	for {
		time.Sleep(1 * time.Second) //每分钟检测一次
		var reSub []NatsSubInfo
		for _, v := range n.clientSubs {
			if v.subscription == nil {
				reSub = append(reSub, v)
				continue
			}
			if !v.subscription.IsValid() {
				v.subscription.Unsubscribe()
				reSub = append(reSub, v)
			}
		}
		// 重新订阅这些消息
		for _, v := range reSub {
			n.NewSubscribe(v)
		}
	}
}
